package com.crazymaker.springcloud.message.service.impl;

import com.crazymaker.springcloud.common.util.ThreadUtil;
import com.crazymaker.springcloud.distribute.zookeeper.ZKClient;
import com.crazymaker.springcloud.standard.context.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.TimeUnit;

import static com.crazymaker.push.platform.contract.constants.PushConstants.WEBSOCKET_SERVER_LOCK_PATH;
import static com.crazymaker.push.platform.contract.constants.PushConstants.WEBSOCKET_SERVER_MONITOR_PATH;
import static com.crazymaker.springcloud.common.util.HttpRequestUtil.simpleGet;
import static com.crazymaker.springcloud.common.util.HttpRequestUtil.simplePost;
import static com.crazymaker.springcloud.message.config.ShardingRuleConfig.DEFAULT_SHARDING_RULE_CONFIG;

/**
 *
 */
@Slf4j
@Component
public class WebSocketServerMonitorService implements ApplicationContextAware {


    @Value("${websocket.register.gateway}")
    private String websocketRegisterGateway;

    public void subscribeDataChanges() {
        boolean nodeOk = ZKClient.getSingleton().isNodeExist(WEBSOCKET_SERVER_MONITOR_PATH);
        if (!nodeOk) {
            ZKClient.getSingleton().createNode(WEBSOCKET_SERVER_MONITOR_PATH, DEFAULT_SHARDING_RULE_CONFIG);
        }
        //注册监听事件

        CuratorFramework client = ZKClient.getSingleton().getClient();


        PathChildrenCache cache =
                new PathChildrenCache(client, WEBSOCKET_SERVER_MONITOR_PATH, true);
        PathChildrenCacheListener l =
                new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client,
                                           PathChildrenCacheEvent event) {
                        try {
                            ChildData data = event.getData();
                            switch (event.getType()) {
                                case CHILD_ADDED:

                                    log.info("子节点增加, path={}, data={}",
                                            data.getPath(), new String(data.getData(), "UTF-8"));

                                    break;
                                case CHILD_UPDATED:
                                    log.info("子节点更新, path={}, data={}",
                                            data.getPath(), new String(data.getData(), "UTF-8"));
                                    break;
                                case CHILD_REMOVED:
                                    log.info("子节点删除, path={}, data={}",
                                            data.getPath(), new String(data.getData(), "UTF-8"));
                                    break;
                                default:
                                    break;
                            }

                            refreshWebsocketNodes();

                        } catch (
                                UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                };
        cache.getListenable().addListener(l);


        //启动节点的事件监听
        try {
            cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public String refreshWebsocketNodes() {
        List<String> list = ZKClient.getSingleton().getChildren(WEBSOCKET_SERVER_MONITOR_PATH);

        if (null == list || list.size() <= 0) {
            log.error("websocket server list is null");
            return null;
        }

        //节点按照编号，升序排列
        Collections.sort(list);
        Iterator<String> iterator = list.iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            String url = "http://" + next.split(":")[0] + ":7704/websocket-provider/api/user/hello?name=foo";
            try {
                String responce = simpleGet(url);
                if (null != responce && StringUtils.isNotEmpty(responce)) {
                    continue;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            iterator.remove();
        }

        if (null == list || list.size() <= 0) {
            log.error("websocket server list is null");
            return null;
        }

        log.debug("nodes is : {}", list);
        log.info("begin to refresh : {}", websocketRegisterGateway);

        InterProcessMutex lock = new InterProcessMutex(ZKClient.getSingleton().getClient(),
                WEBSOCKET_SERVER_LOCK_PATH);
        try {
            lock.acquire();
            Map<String, String> params = new HashMap<>();
            int index = 0;
            for (String item : list) {
                params.put("server"+index++, item);
            }
            params.put("serverCount", String.valueOf(index));

            String url = websocketRegisterGateway;
            String res = simplePost(url, params);
            log.info("websocket gateway {} update ok,{}", websocketRegisterGateway, res);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                lock.release();
            } catch (Exception e) {
                e.printStackTrace();
            }

        }

        return websocketRegisterGateway;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.setContext(applicationContext);

        delaySubscribe();
    }

    private void delaySubscribe() {
        ThreadUtil.delayRun(() -> {

            if (ZKClient.getSingleton().isReady()) {
                subscribeDataChanges();
            } else {
                //如果zk没有连接成功，这块做递归
                delaySubscribe();
            }
        }, ZKClient.DEPEND_DELAY, TimeUnit.SECONDS);
    }

}
